package kafka

import (
	"encoding/json"
	"errors"
	"log"
	"time"

	"github.com/spf13/viper"

	"github.com/Shopify/sarama"
)

var asyncProducerList = make(map[string]*asyncProducer)

// GetAsyncProducerClient 从kafka配置项获取brokerConfKey对应的broker列表，返回异步生产者
func GetAsyncProducerClient(brokerConfKey string) (*asyncProducer, error) {
	if _, ok := asyncProducerList[brokerConfKey]; !ok { //第一次来
		brokerList := viper.GetStringSlice("kafka." + brokerConfKey)
		if len(brokerList) <= 0 {
			return nil, errors.New("在配置文件中未发现该项配置")
		}

		if producerClient, err := newAsyncProducerClient(brokerList, func(config *sarama.Config) {
			config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
			config.Producer.Flush.Frequency = 500 * time.Millisecond  // 每0.5s冲洗一批
			config.Producer.Partitioner = sarama.NewRandomPartitioner // 设置生产者的数据写入哪个分区（此处随机设置一个分区）
		}); err == nil {
			asyncProducerList[brokerConfKey] = &asyncProducer{Producer: producerClient}
		} else {
			return nil, err
		}
	}
	return asyncProducerList[brokerConfKey], nil
}

// newAsyncProducerClient 异步
func newAsyncProducerClient(brokers []string, options ...func(*sarama.Config)) (sarama.AsyncProducer, error) {
	kafkaConfig := sarama.NewConfig()
	kafkaConfig.Producer.Return.Successes = true                // 成功交付的消息将在success channel返回
	kafkaConfig.Producer.Compression = sarama.CompressionSnappy // 压缩消息

	for _, option := range options {
		option(kafkaConfig)
	}

	// 连接kafka
	client, err := sarama.NewAsyncProducer(brokers, kafkaConfig)
	if err != nil {
		return nil, err
	}

	go func(client sarama.AsyncProducer) {
		for {
			select {
			case <-client.Successes():
			//case message := <-client.Successes():
			//	log.Printf("发送消息到kafka成功: %v\n", message)
			case err := <-client.Errors():
				log.Printf("发送消息到kafka失败: %v\n", err)
			}
		}
	}(client)

	return client, nil
}

type asyncProducer struct {
	Producer sarama.AsyncProducer
}

func (p *asyncProducer) Send(topic string, value interface{}) error {
	if p.Producer == nil {
		return errors.New("Kafka producer is nil")
	}

	msg, err := json.Marshal(value)
	if err != nil {
		return err
	}

	p.Producer.Input() <- &sarama.ProducerMessage{
		Topic:     topic,
		Value:     sarama.ByteEncoder(msg),
		Timestamp: time.Now(),
	}

	return nil
}

func (p *asyncProducer) close() error {
	return p.Producer.Close()
}
